/*
 * Copyright (C) 2012  Romain DUBOIS
 * 
 * This file is part of NeMoS - Network and Modular Software
 * 
 * NeMoS is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * NeMoS is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with NeMoS.  If not, see <http://www.gnu.org/licenses/>.
 */
package nemos.remote.socket.internal;

import java.beans.EventHandler;
import java.io.EOFException;
import java.io.IOException;
import java.io.InvalidClassException;
import java.io.NotSerializableException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import nemos.remote.rsa.message.data.Message;
import nemos.remote.rsa.message.transport.AbstractTransport;
import nemos.util.log.NemosLogger;

/**
 * SocketTransport : The socket transport implementation.
 * 
 * @author Romain DUBOIS
 */
public class SocketTransport extends AbstractTransport {

    /** The closing message */
    private static final String CLOSE_MSG = "CLOSE"; //$NON-NLS-1$

    /** The listening thread */
    private final Thread mListeningThread;
    /** The sending thread */
    private final Thread mSendingThread;

    /** The underlying socket */
    private final Socket mSocket;

    /** The waiting messages */
    private final BlockingQueue<Serializable> mWaitingMessages;
    /** The logger */
    private final NemosLogger mLogger;

    /** The listened stream */
    private final ObjectInputStream mInput;
    /** The written stream */
    private final ObjectOutputStream mOutput;

    /**
     * Constructor.
     * 
     * @param pSocket
     *            The socket to use
     * @throws IOException
     *             If first contact failed
     */
    SocketTransport(final Socket pSocket) throws IOException {
        this(null, pSocket);
    }

    /**
     * Constructor.
     * 
     * @param pId
     *            The transport ID
     * @param pSocket
     *            The socket to use
     * @throws IOException
     *             If first contact failed
     */
    public SocketTransport(final String pId, final Socket pSocket) throws IOException {
        super(pId);

        mLogger = new NemosLogger(SocketTransport.class);

        mSocket = pSocket;

        mListeningThread = new Thread(EventHandler.create(Runnable.class, this, "listen")); //$NON-NLS-1$
        mSendingThread = new Thread(EventHandler.create(Runnable.class, this, "send")); //$NON-NLS-1$

        mWaitingMessages = new LinkedBlockingQueue<Serializable>();

        mOutput = new ObjectOutputStream(mSocket.getOutputStream());
        mInput = new ObjectInputStream(mSocket.getInputStream());

        // Handle ID
        if (pId == null) {
            setId(mInput.readUTF());
        } else {
            setId(pId);
            mOutput.writeUTF(pId);
        }

        mSendingThread.start();
        mListeningThread.start();
    }

    /**
     * Listen to incoming messages.
     */
    public void listen() {
        try {
            try {
                while (isActive()) {
                    try {
                        final Object lInputObject = mInput.readObject();
                        if (lInputObject instanceof Message) {
                            incomingMessage((Message) lInputObject);
                        } else if (lInputObject instanceof Ping) {
                            if (((Ping) lInputObject).mAsk) {
                                mWaitingMessages.offer(new Ping(false));
                            }
                        } else if (lInputObject == null) {
                            mLogger.error(SocketMessage.NULL_OBJECT);
                        } else if (CLOSE_MSG.equals(lInputObject)) {
                            // Socket is closing
                            super.close();
                        } else {
                            mLogger.error(SocketMessage.INVALID_TYPE, lInputObject.getClass()
                                    .getName());
                        }
                    } catch (final ClassNotFoundException e) {
                        mLogger.error(SocketMessage.INVALID_TYPE, e);
                    }
                }
            } finally {
                if (!mSocket.isInputShutdown()) {
                    mInput.close();
                }
            }
        } catch (final EOFException e) {
            // Ignore : socket closed
        } catch (final IOException e) {
            if (isActive()) {
                mLogger.error(SocketMessage.LISTENING_ERROR, e);
            }
        }
        close();
    }

    /**
     * Ensure messages are sent.
     */
    public void send() {
        try {
            try {
                while (!Thread.interrupted() && isActive()) {
                    Serializable lMessage = mWaitingMessages.poll(500, TimeUnit.MILLISECONDS);
                    if (lMessage == null) {
                        lMessage = new Ping(true);
                    }

                    try {
                        mOutput.writeObject(lMessage);
                    } catch (final InvalidClassException e) {
                        mLogger.error(SocketMessage.INVALID_TYPE, e);
                    } catch (final NotSerializableException e) {
                        mLogger.error(SocketMessage.INVALID_TYPE, e);
                    } catch (final IOException e) {
                        mLogger.error(SocketMessage.SENDING_ERROR, e);
                    }
                }
            } finally {
                if (!mSocket.isOutputShutdown()) {
                    if (mWaitingMessages.contains(CLOSE_MSG)) {
                        mWaitingMessages.clear();
                        mOutput.writeObject(CLOSE_MSG);
                    }
                    mOutput.close();
                }
            }
        } catch (final InterruptedException e) {
            // Thread interrupted : stop sending
        } catch (final IOException e) {
            if (isActive()) {
                mLogger.error(SocketMessage.SOCKET_ERROR, e);
            }
        }
        close();
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.AbstractTransport#close()
     */
    @Override
    public void close() {
        // First to close
        if (isActive()) {
            mWaitingMessages.offer(CLOSE_MSG);
        }
        super.close();

        // Wait a little
        mSendingThread.interrupt();
        try {
            // Sending thread is easier to wake up
            mSendingThread.join(100);
            // Listening thread must wait for remote close
            mListeningThread.join(1000);
        } catch (final InterruptedException e) {
            // Does not matter : Thread is dying
        }

        // Finally close the socket
        try {
            mSocket.close();
        } catch (final IOException e) {
            mLogger.error(SocketMessage.SOCKET_ERROR, e);
        }
    }

    /**
     * {@inheritDoc}
     * 
     * @see nemos.remote.rsa.message.transport.AbstractTransport#sendMessage(nemos.remote.rsa.message.data.Message)
     */
    @Override
    protected void sendMessage(final Message pMessage) {
        mWaitingMessages.add(pMessage);
    }

    /**
     * Ping Class.
     */
    private static class Ping implements Serializable {

        /** Serial version */
        private static final long serialVersionUID = 1L;

        /** The ask flag */
        final boolean mAsk;

        /**
         * Constructor.
         * 
         * @param pAsk
         *            The ask flag
         */
        Ping(final boolean pAsk) {
            super();
            mAsk = pAsk;
        }

        @Override
        public String toString() {
            if (mAsk) {
                return "Ping"; //$NON-NLS-1$
            }
            return "Pong"; //$NON-NLS-1$
        }
    }
}
